package se.sics.drsm.simulation;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import se.sics.drsm.bootstrap.server.component.DRSMBootstrapServer;
import se.sics.drsm.components.DRSMNode;
import se.sics.drsm.inits.NodeInit;
import se.sics.drsm.model.SONConfiguration;
import se.sics.drsm.overlay_address.key.NumericRingKey;
import se.sics.drsm.ports.drsm.NodePort;
import se.sics.drsm.ports.drsm.events.DRSMJoinRing;
import se.sics.drsm.simulation.commands.PeerFail;
import se.sics.drsm.simulation.commands.PeerJoin;
import se.sics.drsm.simulation.events.DRSMSimulatorInit;
import se.sics.drsm.simulation.events.KilledPeerFault;
import se.sics.drsm.simulation.ports.DRSMSimulationPort;
import se.sics.kompics.*;
import se.sics.kompics.address.Address;
import se.sics.kompics.network.Message;
import se.sics.kompics.network.Network;
import se.sics.kompics.p2p.bootstrap.BootstrapConfiguration;
import se.sics.kompics.p2p.bootstrap.server.BootstrapServerInit;
import se.sics.kompics.p2p.fd.ping.PingFailureDetectorConfiguration;
import se.sics.kompics.p2p.monitor.chord.server.ChordMonitorConfiguration;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.web.Web;
import se.sics.kompics.web.WebRequest;

import java.util.HashMap;
import java.util.Random;

public class DRSMSimulator extends ComponentDefinition {

    Positive<DRSMSimulationPort> simulator = positive(DRSMSimulationPort.class);
    Positive<Network> network = positive(Network.class);
    Positive<Timer> timer = positive(Timer.class);
    Negative<Web> web = negative(Web.class);

    private static final Logger logger = LoggerFactory
            .getLogger(DRSMSimulator.class);

    private final HashMap<Long, Component> peers;
    private final HashMap<Long, Address> peersAddress;

	/* for measuring if the system has stayed in a converged
     * state long enough to stop the simulation */
//	private boolean converged = false;
//	private long convergedAt = 0;

    // peer initialization state
    private Address peer0Address;
    private PingFailureDetectorConfiguration fdConfiguration;
    private BootstrapConfiguration bconfiguration; // Should be default values
    private SONConfiguration sonConfiguration;
    private PingFailureDetectorConfiguration failureDetectorConfiguration = null;
    private ChordMonitorConfiguration drsmMonitorConfiguration;

    private int peerIdSequence;
    private Random random;

    //TODO: Understand the purpose of it
//	private ConsistentHashtable<Long> chordRingView;
//	private Hashtable<Long, Long> chordRingView;

    private Component bootServer;

    public DRSMSimulator() {
        peers = new HashMap<Long, Component>();
        peersAddress = new HashMap<Long, Address>();
//		chordRingView = new Hashtable<Long, Long>();

        subscribe(handleInit, control);

        subscribe(handleJoin, simulator);
//		subscribe(handleFail, simulator);
    }

    Handler<DRSMSimulatorInit> handleInit = new Handler<DRSMSimulatorInit>() {
        public void handle(DRSMSimulatorInit init) {
            bootServer = create(DRSMBootstrapServer.class);
            connect(bootServer.getNegative(Network.class), network);

            peers.clear();
            peerIdSequence = 0;
            random = init.random;

            peer0Address = init.address;
            bconfiguration = init.bconfiguration;
            sonConfiguration = init.chordConfiguration; //TODO change the name
            failureDetectorConfiguration = init.failureDetectorConfiguration;
            drsmMonitorConfiguration = init.chordMonitorConfiguration; // TODO change the name

            // initializing components
            trigger(new BootstrapServerInit(bconfiguration), bootServer
                    .getControl());
        }
    };

    Handler<PeerJoin> handleJoin = new Handler<PeerJoin>() {
        public void handle(PeerJoin event) {
            Long key = event.nodeKey;

            // join with the next id if this id is taken
//			Long successor = chordRingView.getKeySuccessor(key);
//			while (successor != null && successor.equals(key)) {
//				key = (key + 1) % cfg.ID_SIZE;
//				successor = chordRingView.getKeySuccessor(key);
//			}

            logger.debug("JOIN@{}", key);

//			chordRingView.addNode(key);
            createAndStartNewPeer(key);
//			Metrics.increment("Joins");
        }
    };

    Handler<PeerFail> handleFail = new Handler<PeerFail>() {
        public void handle(PeerFail event) {
//			if (chordRingView.size() == 0) {
//				System.err.println("Empty network");
//				return;
//			}
//
//			Long key = chordRingView.getKeySuccessor(event.nodeKey);
//
//			logger.debug("FAIL@" + key);
//
//			chordRingView.remove(key);
//			GlobalView.failed(key);
//
//			stopAndDestroyPeer(key);
//			Metrics.increment("Fails");
        }
    };

    Handler<KilledPeerFault> handleKilledPeerFault = new Handler<KilledPeerFault>() {
        public void handle(KilledPeerFault event) {
//			chordRingView.remove(event.peer.getId());
            stopAndDestroyPeer((long) event.peer.getId());
        }
    };

//	private boolean stableLongEnough() {
//		if (GlobalView.isValid()) {
//			if (!converged) {
//				converged = true;
//				convergedAt = System.currentTimeMillis();
//				return false;
//			}
//			// the system has converged, and was converged last time i checked as well
//			// check for how long it has remained converged
//			if (System.currentTimeMillis()-convergedAt > 5 * cfg.STAB_PERIOD)
//				return true;
//		} else {
//			converged = false;
//			convergedAt = System.currentTimeMillis(); // not needed, should work without it as well
//		}
//		
//		return false;
//	}

    private final Component createAndStartNewPeer(Long nodeKey) {
        Component node = create(DRSMNode.class);
        int peerId = ++peerIdSequence;
        Address peerAddress = new Address(peer0Address.getIp(),
                peer0Address.getPort(), peerId);

//		Negative<NodePort> nodePort = negative(NodePort.class);

        connect(network, node.getNegative(Network.class),
                new MessageDestinationFilter(peerId));

        connect(timer, node.getNegative(Timer.class));
        connect(web, node.getPositive(Web.class),
                new WebRequestDestinationFilter(peerId));

        subscribe(handleKilledPeerFault, node.control());

        // firing inits
        NodeInit nodeInit = new NodeInit(peerAddress, bconfiguration, drsmMonitorConfiguration, sonConfiguration,
                failureDetectorConfiguration);
        trigger(nodeInit, node.getControl());

        DRSMJoinRing joinEvent = new DRSMJoinRing(new NumericRingKey(nodeKey));
        trigger(joinEvent, node.getPositive(NodePort.class));

        trigger(new Start(), node.getControl());
//		peers.put(nodeKey, node);
//		peersAddress.put(nodeKey, new Address(peerAddress, nodeKey));

        // Snapshot.addPeer(peersAddress.get(nodeKey));

        return node;
    }

    private final void stopAndDestroyPeer(Long nodeKey) {
        Component peer = peers.get(nodeKey);

        trigger(new Stop(), peer.getControl());

        unsubscribe(handleKilledPeerFault, peer.control());

        disconnect(network, peer.getNegative(Network.class));
        disconnect(timer, peer.getNegative(Timer.class));
        disconnect(web, peer.getPositive(Web.class));

        peers.remove(nodeKey);
        destroy(peer);

        // Snapshot.removePeer(peersAddress.remove(nodeKey));
    }

    private final static class MessageDestinationFilter extends
            ChannelFilter<Message, Integer> {
        public MessageDestinationFilter(Integer addressId) {
            super(Message.class, addressId, true);
        }

        public Integer getValue(Message event) {
            return event.getDestination().getId();
        }
    }

    private final static class WebRequestDestinationFilter extends
            ChannelFilter<WebRequest, Integer> {
        public WebRequestDestinationFilter(Integer destination) {
            super(WebRequest.class, destination, false);
        }

        public Integer getValue(WebRequest event) {
            return event.getDestination();
        }
    }
}
